package com.iamdigger.queldorei.registry.zookeeper;

import com.iamdigger.queldorei.registry.ServiceAddressSynchronizer;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * zookeeper address synchronizer
 * <p>
 * Created by Sam on 4/8/16.
 *
 * @author Sam Tsai
 * @since 0.1.0
 */
public class ZkServiceAddressSynchronizer extends ServiceAddressSynchronizer implements IZkChildListener {

    private static Logger logger = LoggerFactory.getLogger(ZkServiceAddressSynchronizer.class);

    /**
     * set by service zk discorvery
     */
    private ZkClient zkClient;

    /**
     * zookeeper的服务器地址以及注册目录从discovery获取。
     * 减少配置口子，防止服务主要信息配置重复出现错误。
     */
    private String zkServerAddress;  // zookeeper服务器地址
    private String zkRegistryPath; // zookeeper注册地址

    private int syncSessionTimeout; // 同步服务使用的session超时时间
    private int syncConnectionTimeout; // 同步服务使用的连接超时时间

    @SuppressWarnings("unused")
    public ZkServiceAddressSynchronizer() {
        super();
        this.syncConnectionTimeout = 1000;
        this.syncSessionTimeout = 5000;
    }

    @SuppressWarnings("unused")
    public ZkServiceAddressSynchronizer(int serviceCount, int concurrentLevel,
                                        int sessionTimeout, int connectionTimeout) {
        super(serviceCount, concurrentLevel);
        setupTimeout(sessionTimeout, connectionTimeout);
    }

    private void setupTimeout(int syncSessionTimeout, int syncConnectionTimeout) {
        this.syncSessionTimeout = syncSessionTimeout;
        this.syncConnectionTimeout = syncConnectionTimeout;
    }

    @Override
    public void syncServiceAddress() {
        if (logger.isInfoEnabled()) {
            logger.info("Zookeeper service address synchronizer started. zk server info {}", zkServerAddress);
        }
        if (!zkClient.exists(zkRegistryPath)) {
            // zk不存在根节点，客户端主动创建。
            if (logger.isInfoEnabled()) {
                logger.info("Create non-existed root node {}", zkRegistryPath);
            }
            zkClient.createPersistent(zkRegistryPath);
        }
        zkClient.subscribeChildChanges(zkRegistryPath, this);
        List<String> serviceList = zkClient.getChildren(zkRegistryPath);
        if (null != serviceList && serviceList.size() > 0) {
            for (String serviceName : serviceList) {
                zkClient.subscribeChildChanges(String.format("%s/%s", zkRegistryPath, serviceName), this);
            }
        }
    }

    /**
     * 为了服务的同步, 初始化服务池时拒绝其他线程读取.
     */
    @Override
    public void syncServiceAddressAtStartup() {
        if (!zkClient.exists(zkRegistryPath)) {
            zkClient.createPersistent(zkRegistryPath);
        } else {
            Map<String, List<String>> zkServiceHolder = loadAllZkServices();
            if (null != zkServiceHolder && zkServiceHolder.size() > 0) {
                for (Map.Entry<String, List<String>> entry : zkServiceHolder.entrySet()) {
                    refreshServiceAddress(entry.getKey(), entry.getValue());
                }
            }
        }
    }

    /**
     * 调用之前,请确保zookeeper中存在服务根节点.
     *
     * @return zookeeper中已注册的服务列表
     */
    private Map<String, List<String>> loadAllZkServices() {
        List<String> services = zkClient.getChildren(zkRegistryPath);
        if (null != services && !services.isEmpty()) {
            Map<String, List<String>> srvHolder = new HashMap<String, List<String>>(services.size());
            List<String> addresses;
            for (String serviceName : services) {
                addresses = new ArrayList<String>();
                List<String> addressKeys = zkClient.getChildren(zkRegistryPath + "/" + serviceName);
                if (null == addressKeys || addressKeys.size() == 0) {
                    srvHolder.put(serviceName, addresses);
                    continue;
                }
                for (String addressKey : addressKeys) {
                    String address = zkClient.readData(zkRegistryPath + "/" + serviceName + "/" + addressKey);
                    addresses.add(address);
                }
                srvHolder.put(serviceName, addresses);
            }
            return srvHolder;
        }
        return new HashMap<String, List<String>>();
    }

    public void prepareToSync(String zkServerAddress, String zkRegistryPath) {
        this.zkRegistryPath = zkRegistryPath;
        this.zkServerAddress = zkServerAddress;
        this.zkClient = new ZkClient(zkServerAddress, this.syncSessionTimeout, this.syncConnectionTimeout);
        // 应用刚启动时读取服务地址
        syncServiceAddressAtStartup();
        // 初始化zk客户端之后，开始同步服务地址
        syncServiceAddress();
    }

    @Override
    public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
        //logger.info("\n##zk node watcher## \n{} \n{}", parentPath, currentChilds.toString());
        if (zkRegistryPath.equals(parentPath)) {
            List<String> serviceNames = new ArrayList<String>(currentChilds.size());
            refreshServiceName(serviceNames);
        }

        if (parentPath.startsWith(zkRegistryPath + "/")) {
            List<String> addressList = new ArrayList<String>(currentChilds.size());
            for (String currentChild : currentChilds) {
                addressList.add((String) zkClient.readData(parentPath + "/" + currentChild));
            }
            refreshServiceAddress(parentPath.substring(parentPath.lastIndexOf('/') + 1, parentPath.length()), addressList);
        }
        //if()
        //logger.info("\n====\n{}\n", discoveredServices.toString());
        //logger.info("\n##zk node watcher## \n{} \n{}", parentPath, currentChilds.toString());
    }

    @Override
    public void restoreServiceAddress() {
        //// TODO: 4/18/16 实现订阅异常时对客户端服务池的再次刷新
    }
}
